package major;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.time.Duration;
import java.util.Arrays;

public class ObserverZhang {

    public static void main(String[] args) {
        Consumer<String, String> consumer = new ConsumerBuilder(
                "XiaoZhang", Arrays.asList("blue")).build();
        Producer<String, String> producer = new Observed.ProducerBuiler("xiao_zhang").build();
        Integer xzAge = 0;
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
            for (ConsumerRecord<String, String> record : records) {
                Integer age;
                try{
                    age=Integer.parseInt(record.value());
                }catch (Exception e){
                    age=18;
                }
                xzAge=age+8;
                String k = String.format("妹妹今年%d岁，小张哥哥今年%d岁", age,xzAge);
                System.out.println(k);
                producer.send(new ProducerRecord<>("xiaozhang", xzAge + ""), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        System.out.println(Thread.currentThread().getName()+" --- "+recordMetadata.topic());
                    }
                });
                producer.flush();

            }
        }
    }
}
